/*
 * brin_minmax.c
 *		BRIN 的最小/最大操作类的实现
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/access/brin/brin_minmax.c
 */
#include "postgres.h"

#include "access/brin_internal.h"
#include "access/brin_tuple.h"
#include "access/genam.h"
#include "access/stratnum.h"
#include "catalog/pg_amop.h"
#include "catalog/pg_type.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/syscache.h"

typedef struct MinmaxOpaque
{
	Oid			cached_subtype;
	FmgrInfo	strategy_procinfos[BTMaxStrategyNumber];
} MinmaxOpaque;

static FmgrInfo *fc_minmax_get_strategy_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno,
											  Oid fc_subtype, uint16 fc_strategynum);


Datum brin_minmax_opcinfo(PG_FUNCTION_ARGS)
{
	Oid			fc_typoid = PG_GETARG_OID(0);
	BrinOpcInfo *fc_result;

	/*
	 * opaque->strategy_procinfos 是懒初始化的；在这里，通过 palloc0
	 * 将其设置为全未初始化，fn_oid 被设置为 InvalidOid。
	 */

	fc_result = palloc0(MAXALIGN(SizeofBrinOpcInfo(2)) +
					 sizeof(MinmaxOpaque));
	fc_result->oi_nstored = 2;
	fc_result->oi_regular_nulls = true;
	fc_result->oi_opaque = (MinmaxOpaque *)
		MAXALIGN((char *) fc_result + SizeofBrinOpcInfo(2));
	fc_result->oi_typcache[0] = fc_result->oi_typcache[1] =
		lookup_type_cache(fc_typoid, 0);

	PG_RETURN_POINTER(fc_result);
}

/*
 * 通过将给定的索引元组（它包含某个页面范围的部分状态）与来自另一个
 * 堆元组的给定值进行比较来检查它。如果新值超出了现有元组值指定的最小/最大
 * 范围，则更新索引元组并返回 true。否则，返回 false，并且在这种情况下不
 * 进行修改。
 */
Datum brin_minmax_add_value(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	Datum		fc_newval = PG_GETARG_DATUM(2);
	bool		fc_isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_DATUM(3);
	Oid			fc_colloid = PG_GET_COLLATION();
	FmgrInfo   *fc_cmpFn;
	Datum		fc_compar;
	bool		fc_updated = false;
	Form_pg_attribute fc_attr;
	AttrNumber	fc_attno;

	Assert(!fc_isnull);

	fc_attno = fc_column->bv_attno;
	fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);

	/*
	 * 如果记录的值为 null，将新值（我们知道不是 null）存储为最小值和
	 * 最大值，这样就完成了。
	 */
	if (fc_column->bv_allnulls)
	{
		fc_column->bv_values[0] = datumCopy(fc_newval, fc_attr->attbyval, fc_attr->attlen);
		fc_column->bv_values[1] = datumCopy(fc_newval, fc_attr->attbyval, fc_attr->attlen);
		fc_column->bv_allnulls = false;
		PG_RETURN_BOOL(true);
	}

	/*
	 * 否则，需要将新值与现有边界进行比较并相应地更新。首先检查它是否
	 * 小于现有的最小值。
	 */
	fc_cmpFn = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
										 BTLessStrategyNumber);
	fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_newval, fc_column->bv_values[0]);
	if (DatumGetBool(fc_compar))
	{
		if (!fc_attr->attbyval)
			pfree(DatumGetPointer(fc_column->bv_values[0]));
		fc_column->bv_values[0] = datumCopy(fc_newval, fc_attr->attbyval, fc_attr->attlen);
		fc_updated = true;
	}

	/*
	 * 现在将其与现有的最大值进行比较。
	 */
	fc_cmpFn = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
										 BTGreaterStrategyNumber);
	fc_compar = FunctionCall2Coll(fc_cmpFn, fc_colloid, fc_newval, fc_column->bv_values[1]);
	if (DatumGetBool(fc_compar))
	{
		if (!fc_attr->attbyval)
			pfree(DatumGetPointer(fc_column->bv_values[1]));
		fc_column->bv_values[1] = datumCopy(fc_newval, fc_attr->attbyval, fc_attr->attlen);
		fc_updated = true;
	}

	PG_RETURN_BOOL(fc_updated);
}

/*
 * 给定一个与某个页面范围对应的索引元组和一个扫描键，返回扫描键
 * 是否与索引元组的最小/最大值一致。如果是，则返回 true，否则返回 false。
 *
 * 我们不再处理一致性函数中的 NULL 键，这由 AM 代码处理。这意味着
 * 我们也不应该得到任何全 NULL 范围，因为这些范围无法与常规（非 [IS]
 * NULL）键一致。
 */
Datum brin_minmax_consistent(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_column = (BrinValues *) PG_GETARG_POINTER(1);
	ScanKey		fc_key = (ScanKey) PG_GETARG_POINTER(2);
	Oid			fc_colloid = PG_GET_COLLATION(),
				fc_subtype;
	AttrNumber	fc_attno;
	Datum		fc_value;
	Datum		fc_matches;
	FmgrInfo   *fc_finfo;

	/* 此操作类使用仅有三个参数的旧签名。 */
	Assert(PG_NARGS() == 3);

	/* 不应处理全 NULL 范围。 */
	Assert(!fc_column->bv_allnulls);

	fc_attno = fc_key->sk_attno;
	fc_subtype = fc_key->sk_subtype;
	fc_value = fc_key->sk_argument;
	switch (fc_key->sk_strategy)
	{
		case BTLessStrategyNumber:
		case BTLessEqualStrategyNumber:
			fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
												 fc_key->sk_strategy);
			fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_column->bv_values[0],
										fc_value);
			break;
		case BTEqualStrategyNumber:

			/*
			 * 在相等情况下（WHERE col = someval），如果范围内的最小值
			 * <= 扫描键，并且最大值 >= 扫描键，则返回当前页面范围。
			 */
			fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
												 BTLessEqualStrategyNumber);
			fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_column->bv_values[0],
										fc_value);
			if (!DatumGetBool(fc_matches))
				break;
			/* max() >= scankey */
			fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
												 BTGreaterEqualStrategyNumber);
			fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_column->bv_values[1],
										fc_value);
			break;
		case BTGreaterEqualStrategyNumber:
		case BTGreaterStrategyNumber:
			fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_subtype,
												 fc_key->sk_strategy);
			fc_matches = FunctionCall2Coll(fc_finfo, fc_colloid, fc_column->bv_values[1],
										fc_value);
			break;
		default:
			/* 不应该发生 */
			elog(ERROR, "invalid strategy number %d", fc_key->sk_strategy);
			fc_matches = 0;
			break;
	}

	PG_RETURN_DATUM(fc_matches);
}

/*
 * 给定两个 BrinValues，将第一个值更新为包含两个值中的汇总值的
 * 并集。第二个值保持不变。
 */
Datum brin_minmax_union(PG_FUNCTION_ARGS)
{
	BrinDesc   *fc_bdesc = (BrinDesc *) PG_GETARG_POINTER(0);
	BrinValues *fc_col_a = (BrinValues *) PG_GETARG_POINTER(1);
	BrinValues *fc_col_b = (BrinValues *) PG_GETARG_POINTER(2);
	Oid			fc_colloid = PG_GET_COLLATION();
	AttrNumber	fc_attno;
	Form_pg_attribute fc_attr;
	FmgrInfo   *fc_finfo;
	bool		fc_needsadj;

	Assert(fc_col_a->bv_attno == fc_col_b->bv_attno);
	Assert(!fc_col_a->bv_allnulls && !fc_col_b->bv_allnulls);

	fc_attno = fc_col_a->bv_attno;
	fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);

	/* 如果 B 的最小值小于 A 的最小值，则调整最小值 */
	fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
										 BTLessStrategyNumber);
	fc_needsadj = FunctionCall2Coll(fc_finfo, fc_colloid, fc_col_b->bv_values[0],
								 fc_col_a->bv_values[0]);
	if (fc_needsadj)
	{
		if (!fc_attr->attbyval)
			pfree(DatumGetPointer(fc_col_a->bv_values[0]));
		fc_col_a->bv_values[0] = datumCopy(fc_col_b->bv_values[0],
										fc_attr->attbyval, fc_attr->attlen);
	}

	/* 如果 B 的最大值大于 A 的最大值，则调整最大值 */
	fc_finfo = fc_minmax_get_strategy_procinfo(fc_bdesc, fc_attno, fc_attr->atttypid,
										 BTGreaterStrategyNumber);
	fc_needsadj = FunctionCall2Coll(fc_finfo, fc_colloid, fc_col_b->bv_values[1],
								 fc_col_a->bv_values[1]);
	if (fc_needsadj)
	{
		if (!fc_attr->attbyval)
			pfree(DatumGetPointer(fc_col_a->bv_values[1]));
		fc_col_a->bv_values[1] = datumCopy(fc_col_b->bv_values[1],
										fc_attr->attbyval, fc_attr->attlen);
	}

	PG_RETURN_VOID();
}

/*
 * 缓存并返回给定策略的过程。
 *
 * 注意：此函数镜像了 inclusion_get_strategy_procinfo；请查看
 * 那里的注释。如果在这里进行更改，也请查看该函数。
 */
static FmgrInfo * fc_minmax_get_strategy_procinfo(BrinDesc *fc_bdesc, uint16 fc_attno, Oid fc_subtype,
							 uint16 fc_strategynum)
{
	MinmaxOpaque *fc_opaque;

	Assert(fc_strategynum >= 1 &&
		   fc_strategynum <= BTMaxStrategyNumber);

	fc_opaque = (MinmaxOpaque *) fc_bdesc->bd_info[fc_attno - 1]->oi_opaque;

	/*
	 * 我们在 opaque 结构中缓存之前子类型的过程，以避免重复的 syscache
	 * 查找。如果子类型发生更改，则使所有缓存条目无效。
	 */
	if (fc_opaque->cached_subtype != fc_subtype)
	{
		uint16		fc_i;

		for (fc_i = 1; fc_i <= BTMaxStrategyNumber; fc_i++)
			fc_opaque->strategy_procinfos[fc_i - 1].fn_oid = InvalidOid;
		fc_opaque->cached_subtype = fc_subtype;
	}

	if (fc_opaque->strategy_procinfos[fc_strategynum - 1].fn_oid == InvalidOid)
	{
		Form_pg_attribute fc_attr;
		HeapTuple	fc_tuple;
		Oid			fc_opfamily,
					fc_oprid;
		bool		fc_isNull;

		fc_opfamily = fc_bdesc->bd_index->rd_opfamily[fc_attno - 1];
		fc_attr = TupleDescAttr(fc_bdesc->bd_tupdesc, fc_attno - 1);
		fc_tuple = SearchSysCache4(AMOPSTRATEGY, ObjectIdGetDatum(fc_opfamily),
								ObjectIdGetDatum(fc_attr->atttypid),
								ObjectIdGetDatum(fc_subtype),
								Int16GetDatum(fc_strategynum));

		if (!HeapTupleIsValid(fc_tuple))
			elog(ERROR, "missing operator %d(%u,%u) in opfamily %u",
				 fc_strategynum, fc_attr->atttypid, fc_subtype, fc_opfamily);

		fc_oprid = DatumGetObjectId(SysCacheGetAttr(AMOPSTRATEGY, fc_tuple,
												 Anum_pg_amop_amopopr, &fc_isNull));
		ReleaseSysCache(fc_tuple);
		Assert(!fc_isNull && RegProcedureIsValid(fc_oprid));

		fmgr_info_cxt(get_opcode(fc_oprid),
					  &fc_opaque->strategy_procinfos[fc_strategynum - 1],
					  fc_bdesc->bd_context);
	}

	return &fc_opaque->strategy_procinfos[fc_strategynum - 1];
}
